#####################################################################
# This source file contains functions common to both master and slave
# nodes. basically a bunch of functions to abstract MPI primitives
# (instead of calling a bunch of primitives on a node, we 
#  encapsulate it into a couple of simpler function calls)
#

####################################################################
# wrapper for each slave to process intended data w/ intended function
docall <- function(fun, args) {
    #wrap given function
    if ((is.character(fun) && length(fun) == 1) || is.name(fun))
        fun <- get(as.character(fun), envir = .GlobalEnv, mode="function")
    #execute function using its args
    do.call("fun", lapply(args,enquote))
}

#####################################################################
# RMPI primitive wrappers
#
sendData <- function(node, data) {
    mpi.send.Robj(data, node$rank, node$SENDTAG, node$comm)
}

recvData <- function(node) {
    mpi.recv.Robj(node$rank, node$RECVTAG, node$comm)
}

# wraps up the data and the message tags
postData <- function(node, type, value=NULL, tag=NULL) {
    sendData(node, list(type=type, data=value, tag=tag))
}

#####################################################################
# this function should be run by the master. it waits for any slave node
# to be complete in processing its data, then returns the result to
# the master
recvIData <- function(cl) {
    comm <- cl[[1]]$comm
    status <- 0
    
    # figure out receive tag
#    cat("Figure out receive tag...\n")
    rtag <- NULL
    for (node in cl) {
        if (is.null(rtag))
            rtag <- node$RECVTAG
        else if (rtag != node$RECVTAG) {
            rtag <- mpi.any.tag()
            break;
        }
    }

    # figure out where a message is coming from
    # mpi probe is a blocking call - waits for a message to come
#    cat(paste("Receiving tag is ",rtag,". Waiting for message...\n"))
    mpi.probe(mpi.any.source(), rtag, comm, status)
    
#    cat("Received status of incoming message. Now receiving the message...\n")
    srctag <- mpi.get.sourcetag(status)
    data <- mpi.recv.Robj(srctag[1], srctag[2], comm)
#    cat("Message received!\n")
    list(node = srctag[1], value=data)
}

####################################################################
# Node data structs
# Note that slaves only communicate with their master
#
newSlave <- function(r, c)
    structure(list(rank=r, RECVTAG=33, SENDTAG=22, comm=c))

newMaster <- function(c)
    structure(list(rank=0, RECVTAG=22, SENDTAG=33, comm=c))

#define the master node
#comm is always 1
master <- newMaster(1)

##########################################################################
# Intended only to be run by slave nodes. Repeatedly query for new messages
# sent by the master node. Once it receives a message, it operates on that
# message's data using the function and function args along with it.
# This is the meat of our message passing infrastructure in R
recvQuery <- function(master) {
    repeat tryCatch({
#        cat("Starting query receive loop...\n")
        # this is a blocking receive, the slave nodes should be
        # stalled here until a message arrives from the master
        msg <- recvData(master)

        if (msg$type == "DONE") {
            # we're done, tell the slave to break out of the loop
            break;
        }
        else if (msg$type == "EXEC") {
            success <- TRUE

            #handle errors
            handler <- function(e) {
                success <<- FALSE
                structure(conditionMessage(e))
            }

            #do work
#            cat("Rank ", mpi.comm.rank(), "trying to do call.\n")
            t1 <- proc.time()
            value <- tryCatch(docall(msg$data$fun, msg$data$args),
                                     error=handler)
            t2 <- proc.time()

            #return results back to master
#            cat("Rank ", mpi.comm.rank(), "is done. Returning...\n")
            value <- list(type="VALUE", value=value, success=success,
                          time=t2-t1, tag=msg$data$tag)
            sendData(master, value)
        }
    }, interrupt = function(e) NULL)
}

###################################################################
# Some higher level functions
recvIResult <- function(cl) {

    #wrap call with timing functions
    if (.timing$running()) {
        start <- proc.time()[3]
#        cat("In recvIResult...\n")
        v <- recvIData(cl)
        end <- proc.time()[3]
        .timing$enterRecv(v$node, start, end, v$value$time[3])
    }
    else v <- recvIData(cl)

    #return results
    list(value=v$value$value, node=v$node, tag=v$value$tag)
}

sendCall <- function(node, fun, args, return=TRUE, tag=NULL) {
    timing <- .timing$running()
    if (timing) start <- proc.time()[3]
    postData(node, "EXEC", list(fun=fun, args=args, return=return, tag=tag))
    if (timing) .timing$enterSend(node$rank, start, proc.time()[3])    
}
